package juc.executors;

import java.util.Collection;
import java.util.List;
import java.util.concurrent.*;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

public class ExecutorServiceExample4 {

    public static void main(String[] args) throws ExecutionException, InterruptedException, TimeoutException {
        //testInvokeAny();
        //testInvokeAll();
        //testSubmitRunnable();
        testSubmitRunnableWithResult();
    }

    /**
     * {@link ExecutorService#invokeAny(Collection, long, TimeUnit)}
     * Question:
     * when the result returned,other callable will be keep on process?
     * Answer:
     * Other callable will be canceled (if other not process finished).
     */
    private static void testInvokeAny() throws ExecutionException, InterruptedException, TimeoutException {
        ExecutorService executorService = Executors.newFixedThreadPool(10);

        List<Callable<Integer>> callableList = IntStream.range(0, 5).boxed().map(integer ->
                (Callable<Integer>) () -> {
                    TimeUnit.SECONDS.sleep(ThreadLocalRandom.current().nextInt(10));
                    System.out.println(Thread.currentThread().getName() + ":" + integer);
                    return integer;
                }
        ).collect(Collectors.toList());
        //invokeAny()是同步方法，返回其中一个线程的结果
        Integer value = executorService.invokeAny(callableList, 1, TimeUnit.SECONDS);
        System.out.println("finished:" + value);
    }

    /**
     * {@link ExecutorService#invokeAll(Collection, long, TimeUnit)}
     */
    private static void testInvokeAll() throws InterruptedException {
        ExecutorService executorService = Executors.newFixedThreadPool(10);

        List<Callable<Integer>> callableList = IntStream.range(0, 5).boxed().map(integer ->
                (Callable<Integer>) () -> {
                    TimeUnit.SECONDS.sleep(ThreadLocalRandom.current().nextInt(10));
                    System.out.println(Thread.currentThread().getName() + ":" + integer);
                    return integer;
                }
        ).collect(Collectors.toList());
        //同步方法
        List<Future<Integer>> futureList = executorService.invokeAll(callableList);
        //要等全部结果返回之后才能执行，RXjava、jdk9里支持异步
        futureList.parallelStream().map(future -> {
            try {
                return future.get();
            } catch (InterruptedException | ExecutionException e) {
                throw new RuntimeException(e);
            }
        }).forEach(System.out::println);
        System.out.println("finished");
    }

    /**
     * {@link ExecutorService#submit(Runnable)}
     */
    private static void testSubmitRunnable() throws ExecutionException, InterruptedException {
        ExecutorService executorService = Executors.newFixedThreadPool(10);
        //和execute一样是异步方法
        //泛型是个问号，因为submit的是Runnable没有返回值
        Future<?> future = executorService.submit(() -> {
            timeSleep(2);
        });
        //没有返回值get()的是null
        Object NULL = future.get();
        System.out.println("result:" + NULL);
    }

    /**
     * {@link ExecutorService#submit(Runnable, Object)}
     */
    private static void testSubmitRunnableWithResult() throws ExecutionException, InterruptedException {
        ExecutorService executorService = Executors.newFixedThreadPool(10);
        String result = "DONE";
        Future<String> future = executorService.submit(() -> {
            timeSleep(2);
        }, result);
        System.out.println("result:" + future.get());
    }

    private static void timeSleep(long time) {
        try {
            TimeUnit.SECONDS.sleep(time);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
